/**     
 * @Title: AbstractMessageHandler.java   
 * @Package me.xxy.mq   
 * @Description: TODO
 * @author weiwei 
 * @date 2018年5月30日 下午5:03:33   
 * @version V1.0     
 */
package me.xxy.mq.core;

import javax.annotation.PostConstruct;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskRejectedException;

/**
 * @ClassName: AbstractMessageHandler
 * @Description: TODO
 * @author weiwei
 * @date 2018年5月30日 下午5:03:33
 * 
 */
public abstract class AbstractMessageHandler<T> implements Runnable {

	private static Log LOG = LogFactory.getLog(AbstractMessageHandler.class);

	@Autowired
	protected MessageTrunk messageTrunk;

	@Autowired
	private RedisConfig redisConfig;

	@PostConstruct
	public void start() {
		// 启动监听
		new Thread(this).start();
	}

	/** 监听的消息类型 **/
	private String messageType;

	/** 消息的class对象 **/
	@SuppressWarnings("unused")
	private Class<T> clazz;

	private boolean monitor;

	/** 默认为3 **/
	private int retryTimes = 3;

	public AbstractMessageHandler(String messageType, Class<T> clazz, int retryTimes) {
		this.messageType = messageType;
		this.clazz = clazz;
		this.retryTimes = retryTimes;
	}

	public AbstractMessageHandler(String messageType, Class<T> clazz) {
		this.messageType = messageType;
		this.clazz = clazz;
	}

	@Override
	public void run() {
		while (true) {
			listen();
		}
	}

	/**
	 * @Title: listen
	 * @Description: 监听消息
	 * @Reutrn void
	 */
	@SuppressWarnings("unchecked")
	public void listen() {
		LOG.debug("*********** 监听消息 ************");
		LOG.debug("messageType >> " + messageType);
		// 阻塞获取redis列表值
		final Object obj = redisConfig.brpop(messageType, Integer.MAX_VALUE, Message.class);
		LOG.debug("*********** brpop >> " + obj);
		// 如果获取失败,以为着redis连接有问题
		if (null == obj) {
			monitor = false;
			LOG.warn("消息分发器获取redis连接失败");
			// 暂停5秒钟,防止线程空跑,打印无数日志
			try {
				Thread.sleep(5000);
			} catch (InterruptedException e) {
				LOG.warn("消息分发器线程暂停失败");
			}
			return;
		}
		if (!monitor) {
			LOG.warn("消息分发开始");
			monitor = true;
		}

		try {
			messageTrunk.getThreadPool().submit(new Runnable() {
				public void run() {
					Message message = (Message) obj;
					try {
						// 如果获取成功，则交给子类做业务处理
						handle((T) message.getContent());
					} catch (Exception ex) {
						LOG.error(ex);
						// 处理失败，判断是否需要重试
						if (message.getTimes().intValue() >= retryTimes) {
							handleFailed((T) message.getContent());
						} else {
							message.setTimes(message.getTimes().intValue() + 1);
							// 再次put回消息总线，等待下次重试
							messageTrunk.put(message);
							if (LOG.isDebugEnabled()) {
								StringBuilder sb = new StringBuilder();
								sb.append("msg:[").append(message).append("], 执行失败，准备重试。");
								LOG.debug(sb.toString());
							}
						}
					}
				}
			});
		} catch (TaskRejectedException ex) {
			LOG.warn("线程池已满，准备回写任务，暂停本线程");
			// 如果发生任务拒绝加入线程池，则回写任务到redis，等待重试
			messageTrunk.put((Message) obj);
			// 暂停指定时间
			try {
				Thread.sleep(messageTrunk.getThreadPoolFullSleepSeconds() * 1000);
			} catch (InterruptedException e) {
				LOG.warn("生产者暂停异常", ex);
			}
		} catch (Exception ex) {
			LOG.error("消息总线发生异常", ex);
		}
	}

	/**
	 * @Title: handle
	 * @Description: 获取到消息后做业务处理
	 * @param obj
	 * @Reutrn void
	 */
	public abstract void handle(T obj);

	/**
	 * @Title: handleFailed
	 * @Description: 消息多次重试处理失败
	 * @param obj
	 * @Reutrn void
	 */
	public abstract void handleFailed(T obj);
}